package drds.binlog.instance.manager;

import drds.binlog.common.BinlogException;
import drds.binlog.common.alarm.AlarmHandler;
import drds.binlog.common.alarm.LogAlarmHandler;
import drds.binlog.common.position.EntryPosition;
import drds.binlog.common.utils.JsonUtils;
import drds.binlog.common.zookeeper.ZooKeeperClient;
import drds.binlog.filter.aviater.AviaterRegexFilter;
import drds.binlog.instance.core.AbstractBinlogTask;
import drds.binlog.instance.manager.model.BinlogTaskInfo;
import drds.binlog.instance.manager.model.Parameter;
import drds.binlog.instance.manager.model.Parameter.*;
import drds.binlog.metadata.FileMixedMetaDataManager;
import drds.binlog.metadata.MemoryMetaDataManager;
import drds.binlog.metadata.PeriodMixedMetaDataManager;
import drds.binlog.metadata.ZooKeeperMetaDataManager;
import drds.binlog.parse.EventParser;
import drds.binlog.parse.ha.HAController;
import drds.binlog.parse.ha.HeartBeatHAController;
import drds.binlog.parse.inbound.AbstractEventParser;
import drds.binlog.parse.inbound.group.GroupEventParser;
import drds.binlog.parse.inbound.mysql.LocalBinlogEventParser;
import drds.binlog.parse.inbound.mysql.MysqlEventParser;
import drds.binlog.parse.index.*;
import drds.binlog.parse.support.AuthenticationInfo;
import drds.binlog.sink.entry.EntryEventSink;
import drds.binlog.sink.entry.group.GroupEventSink;
import drds.binlog.store.AbstractStoreCleanUp;
import drds.binlog.store.MemoryEventStore;
import drds.binlog.store.model.BatchMode;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * 单个canal实例，比如一个destination会独立一个实例
 */
public class BinlogTaskWithManager extends AbstractBinlogTask
{

    private static final Logger logger = LoggerFactory.getLogger(BinlogTaskWithManager.class);

    protected Parameter parameter;                                                      // 对应参数
    protected String filter;                                                          // 过滤表达式

    public BinlogTaskWithManager(BinlogTaskInfo binlogTaskInfo, String filter)
    {

        this.binlogTaskId = binlogTaskInfo.getBinlogTaskId();
        this.destination = binlogTaskInfo.getDestination();
        this.parameter = binlogTaskInfo.getParameter();
        //
        this.filter = filter;

        logger.info("init BinlogTask for {}-{} with parameter:{}", binlogTaskId, destination, parameter);
        // 初始化报警机制
        initAlarmHandler();
        // 初始化metaManager
        initMetaManager();
        // 初始化eventStore
        initEventStore();
        // 初始化eventSink
        initEventSink();
        // 初始化eventParser;
        initEventParser();

        // 基础工具，需要提前start，会有先订阅再根据filter条件启动parse的需求
        if (!alarmHandler.isRunning())
        {
            alarmHandler.start();
        }

        if (!metaDataManager.isRunning())
        {
            metaDataManager.start();
        }
        logger.info("init successful....");
    }

    public void start()
    {
        // 初始化metaManager
        logger.info("start CannalInstance for {}-{} with parameter:{}", binlogTaskId, destination, parameter);
        super.start();
    }

    protected void initAlarmHandler()
    {
        logger.info("init alarmHandler begin...");
        alarmHandler = new LogAlarmHandler();
        logger.info("init alarmHandler end! \n\t load AlarmHandler:{} ", alarmHandler.getClass().getName());
    }

    protected void initMetaManager()
    {
        logger.info("init metaDataManager begin...");
        MetaDataMode metaDataMode = parameter.getMetaDataMode();
        if (metaDataMode.isMemory())
        {
            metaDataManager = new MemoryMetaDataManager();
        } else if (metaDataMode.isZookeeper())
        {
            metaDataManager = new ZooKeeperMetaDataManager();
            ((ZooKeeperMetaDataManager) metaDataManager).setZooKeeperClient(getZkclientx());
        } else if (metaDataMode.isMixed())
        {
            // metaDataManager = new MixedMetaDataManager();
            metaDataManager = new PeriodMixedMetaDataManager();// 换用优化过的mixed, at
            // 2012-09-11
            // 设置内嵌的zk metaDataManager
            ZooKeeperMetaDataManager zooKeeperMetaManager = new ZooKeeperMetaDataManager();
            zooKeeperMetaManager.setZooKeeperClient(getZkclientx());
            ((PeriodMixedMetaDataManager) metaDataManager).setZooKeeperMetaManager(zooKeeperMetaManager);
        } else if (metaDataMode.isLocalFile())
        {
            FileMixedMetaDataManager fileMixedMetaManager = new FileMixedMetaDataManager();
            fileMixedMetaManager.setDataDir(parameter.getDataDir());
            fileMixedMetaManager.setPeriod(parameter.getMetaFileFlushPeriod());
            metaDataManager = fileMixedMetaManager;
        } else
        {
            throw new BinlogException("unsupport MetaDataMode for " + metaDataMode);
        }

        logger.info("init metaDataManager end! \n\t load MetaDataManager:{} ", metaDataManager.getClass().getName());
    }

    protected void initEventStore()
    {
        logger.info("init eventStore begin...");
        StorageMode storageMode = parameter.getStorageMode();
        if (storageMode.isMemory())
        {
            MemoryEventStore memoryEventStore = new MemoryEventStore();
            memoryEventStore.setEventsSize(parameter.getMemoryStorageBufferSize());
            memoryEventStore.setMemorySizePerEvent(parameter.getMemoryStorageBufferMemUnit());
            memoryEventStore.setBatchMode(BatchMode.valueOf(parameter.getStorageBatchMode().name()));
            memoryEventStore.setDdlIsolation(parameter.getDdlIsolation());
            eventStore = memoryEventStore;
        } else if (storageMode.isFile())
        {
            // 后续版本支持
            throw new BinlogException("unsupport MetaDataMode for " + storageMode);
        } else if (storageMode.isMixed())
        {
            // 后续版本支持
            throw new BinlogException("unsupport MetaDataMode for " + storageMode);
        } else
        {
            throw new BinlogException("unsupport MetaDataMode for " + storageMode);
        }

        if (eventStore instanceof AbstractStoreCleanUp)
        {
            StorageScavengeMode storageScavengeMode = parameter.getStorageScavengeMode();
            AbstractStoreCleanUp eventScavengeStore = (AbstractStoreCleanUp) eventStore;
            eventScavengeStore.setDestination(destination);
            eventScavengeStore.setMetaDataManager(metaDataManager);
            eventScavengeStore.setOnAck(storageScavengeMode.isOnAck());
            eventScavengeStore.setOnFull(storageScavengeMode.isOnFull());
            eventScavengeStore.setOnSchedule(storageScavengeMode.isOnSchedule());
            if (storageScavengeMode.isOnSchedule())
            {
                eventScavengeStore.setScavengeSchedule(parameter.getScavengeSchdule());
            }
        }
        logger.info("init eventStore end! \n\t load EventStore:{}", eventStore.getClass().getName());
    }

    protected void initEventSink()
    {
        logger.info("init eventSink begin...");

        int groupSize = getGroupSize();
        if (groupSize <= 1)
        {
            eventSink = new EntryEventSink();
        } else
        {
            eventSink = new GroupEventSink(groupSize);
        }

        if (eventSink instanceof EntryEventSink)
        {
            ((EntryEventSink) eventSink).setFilterTransactionEntry(false);
            ((EntryEventSink) eventSink).setEventStore(getEventStore());
        }
        // if (StringUtils.isNotEmpty(eventFilter)) {
        // AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(eventFilter);
        // ((AbstractEventSink) eventSink).setEventFilter(aviaterFilter);
        // }
        logger.info("init eventSink end! \n\t load EventSink:{}", eventSink.getClass().getName());
    }

    protected void initEventParser()
    {
        logger.info("init eventParser begin...");
        SourcingType sourcingType = parameter.getSourcingType();

        List<List<DataSourcing>> groupDbAddresses = parameter.getGroupDbAddresses();
        if (!CollectionUtils.isEmpty(groupDbAddresses))
        {
            int size = groupDbAddresses.get(0).size();// 取第一个分组的数量，主备分组的数量必须一致
            List<EventParser> eventParsers = new ArrayList<EventParser>();
            for (int i = 0; i < size; i++)
            {
                List<InetSocketAddress> dbAddress = new ArrayList<InetSocketAddress>();
                SourcingType lastType = null;
                for (List<DataSourcing> groupDbAddress : groupDbAddresses)
                {
                    if (lastType != null && !lastType.equals(groupDbAddress.get(i).getType()))
                    {
                        throw new BinlogException(String.format("master/slave Sourcing type is unmatch. %s vs %s",
                                lastType,
                                groupDbAddress.get(i).getType()));
                    }

                    lastType = groupDbAddress.get(i).getType();
                    dbAddress.add(groupDbAddress.get(i).getDbAddress());
                }

                // 初始化其中的一个分组parser
                eventParsers.add(doInitEventParser(lastType, dbAddress));
            }

            if (eventParsers.size() > 1)
            { // 如果存在分组，构造分组的parser
                GroupEventParser groupEventParser = new GroupEventParser();
                groupEventParser.setEventParserList(eventParsers);
                this.eventParser = groupEventParser;
            } else
            {
                this.eventParser = eventParsers.get(0);
            }
        } else
        {
            // 创建一个空数据库地址的parser，可能使用了tddl指定地址，启动的时候才会从tddl获取地址
            this.eventParser = doInitEventParser(sourcingType, new ArrayList<InetSocketAddress>());
        }

        logger.info("init eventParser end! \n\t load EventParser:{}", eventParser.getClass().getName());
    }

    private EventParser doInitEventParser(SourcingType sourcingType, List<InetSocketAddress> dbAddresses)
    {
        EventParser eventParser;
        if (sourcingType.isMysql())
        {
            MysqlEventParser mysqlEventParser = new MysqlEventParser();
            mysqlEventParser.setDestination(destination);
            // 编码参数
            mysqlEventParser.setConnectionCharset(Charset.forName(parameter.getConnectionCharset()));
            mysqlEventParser.setConnectionCharsetNumber(parameter.getConnectionCharsetNumber());
            // 网络相关参数
            mysqlEventParser.setDefaultConnectionTimeoutInSeconds(parameter.getDefaultConnectionTimeoutInSeconds());
            mysqlEventParser.setSendBufferSize(parameter.getSendBufferSize());
            mysqlEventParser.setReceiveBufferSize(parameter.getReceiveBufferSize());
            // 心跳检查参数
            mysqlEventParser.setDetectingEnable(parameter.getDetectingEnable());
            mysqlEventParser.setDetectingSQL(parameter.getDetectingSQL());
            mysqlEventParser.setDetectingIntervalInSeconds(parameter.getDetectingIntervalInSeconds());
            // 数据库信息参数
            mysqlEventParser.setSlaveId(parameter.getSlaveId());
            if (!CollectionUtils.isEmpty(dbAddresses))
            {
                mysqlEventParser.setMasterInfo(new AuthenticationInfo(dbAddresses.get(0),
                        parameter.getDbUsername(),
                        parameter.getDbPassword(),
                        parameter.getDefaultDatabaseName()));

                if (dbAddresses.size() > 1)
                {
                    mysqlEventParser.setStandbyInfo(new AuthenticationInfo(dbAddresses.get(1),
                            parameter.getDbUsername(),
                            parameter.getDbPassword(),
                            parameter.getDefaultDatabaseName()));
                }
            }

            if (!CollectionUtils.isEmpty(parameter.getPositions()))
            {
                EntryPosition masterPosition = JsonUtils.unmarshalFromString(parameter.getPositions().get(0),
                        EntryPosition.class);
                // binlog位置参数
                mysqlEventParser.setMasterPosition(masterPosition);

                if (parameter.getPositions().size() > 1)
                {
                    EntryPosition standbyPosition = JsonUtils.unmarshalFromString(parameter.getPositions().get(1),
                            EntryPosition.class);
                    mysqlEventParser.setStandbyPosition(standbyPosition);
                }
            }
            mysqlEventParser.setFallbackIntervalInSeconds(parameter.getFallbackIntervalInSeconds());
            mysqlEventParser.setProfilingEnabled(false);
            mysqlEventParser.setFilterTableError(parameter.getFilterTableError());
            eventParser = mysqlEventParser;
        } else if (sourcingType.isLocalBinlog())
        {
            LocalBinlogEventParser localBinlogEventParser = new LocalBinlogEventParser();
            localBinlogEventParser.setDestination(destination);
            localBinlogEventParser.setBufferSize(parameter.getReceiveBufferSize());
            localBinlogEventParser.setConnectionCharset(Charset.forName(parameter.getConnectionCharset()));
            localBinlogEventParser.setConnectionCharsetNumber(parameter.getConnectionCharsetNumber());
            localBinlogEventParser.setDirectory(parameter.getLocalBinlogDirectory());
            localBinlogEventParser.setProfilingEnabled(false);
            localBinlogEventParser.setDetectingEnable(parameter.getDetectingEnable());
            localBinlogEventParser.setDetectingIntervalInSeconds(parameter.getDetectingIntervalInSeconds());
            localBinlogEventParser.setFilterTableError(parameter.getFilterTableError());
            // 数据库信息，反查表结构时需要
            if (!CollectionUtils.isEmpty(dbAddresses))
            {
                localBinlogEventParser.setMasterInfo(new AuthenticationInfo(dbAddresses.get(0),
                        parameter.getDbUsername(),
                        parameter.getDbPassword(),
                        parameter.getDefaultDatabaseName()));

            }
            eventParser = localBinlogEventParser;
        } else if (sourcingType.isOracle())
        {
            throw new BinlogException("unsupport SourcingType for " + sourcingType);
        } else
        {
            throw new BinlogException("unsupport SourcingType for " + sourcingType);
        }

        // add transaction support at 2012-12-06
        if (eventParser instanceof AbstractEventParser)
        {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            abstractEventParser.setTransactionSize(parameter.getTransactionSize());
            abstractEventParser.setLogPositionManager(initLogPositionManager());
            abstractEventParser.setAlarmHandler(getAlarmHandler());
            abstractEventParser.setEventSink(getEventSink());

            if (StringUtils.isNotEmpty(filter))
            {
                AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(filter);
                abstractEventParser.setEventFilter(aviaterFilter);
            }

            // 设置黑名单
            if (StringUtils.isNotEmpty(parameter.getBlackFilter()))
            {
                AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(parameter.getBlackFilter());
                abstractEventParser.setEventBlackFilter(aviaterFilter);
            }
        }
        if (eventParser instanceof MysqlEventParser)
        {
            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;

            // 初始化haController，绑定与eventParser的关系，haController会控制eventParser
            HAController haController = initHaController();
            mysqlEventParser.setHaController(haController);
        }
        return eventParser;
    }

    protected HAController initHaController()
    {
        logger.info("init haController begin...");
        HaMode haMode = parameter.getHaMode();
        HAController haController = null;
        if (haMode.isHeartBeat())
        {
            haController = new HeartBeatHAController();
            ((HeartBeatHAController) haController).setDetectingRetryTimes(parameter.getDetectingRetryTimes());
            ((HeartBeatHAController) haController).setSwitchEnable(parameter.getHeartbeatHaEnable());
        } else
        {
            throw new BinlogException("unsupport HaMode for " + haMode);
        }
        logger.info("init haController end! \n\t load HAController:{}", haController.getClass().getName());

        return haController;
    }

    protected LogPositionManager initLogPositionManager()
    {
        logger.info("init logPositionPersistManager begin...");
        IndexMode indexMode = parameter.getIndexMode();
        LogPositionManager logPositionManager;
        if (indexMode.isMemory())
        {
            logPositionManager = new MemoryLogPositionManager();
        } else if (indexMode.isZookeeper())
        {
            logPositionManager = new ZooKeeperLogPositionManager(getZkclientx());
        } else if (indexMode.isMixed())
        {
            MemoryLogPositionManager memoryLogPositionManager = new MemoryLogPositionManager();
            ZooKeeperLogPositionManager zooKeeperLogPositionManager = new ZooKeeperLogPositionManager(getZkclientx());
            logPositionManager = new PeriodMixedLogPositionManager(memoryLogPositionManager,
                    zooKeeperLogPositionManager,
                    1000L);
        } else if (indexMode.isMeta())
        {
            logPositionManager = new MetaLogPositionManager(metaDataManager);
        } else if (indexMode.isMemoryMetaFailback())
        {
            MemoryLogPositionManager primary = new MemoryLogPositionManager();
            MetaLogPositionManager secondary = new MetaLogPositionManager(metaDataManager);

            logPositionManager = new FailbackLogPositionManager(primary, secondary);
        } else
        {
            throw new BinlogException("unsupport indexMode for " + indexMode);
        }

        logger.info("init logPositionManager end! \n\t load LogPositionManager:{}", logPositionManager.getClass()
                .getName());

        return logPositionManager;
    }

    protected void startEventParserInternal(EventParser eventParser, boolean isGroup)
    {
        if (eventParser instanceof AbstractEventParser)
        {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            abstractEventParser.setAlarmHandler(getAlarmHandler());
        }

        super.startEventParserInternal(eventParser, isGroup);
    }

    private int getGroupSize()
    {
        List<List<DataSourcing>> groupDbAddresses = parameter.getGroupDbAddresses();
        if (!CollectionUtils.isEmpty(groupDbAddresses))
        {
            return groupDbAddresses.get(0).size();
        } else
        {
            // 可能是基于tddl的启动
            return 1;
        }
    }

    private synchronized ZooKeeperClient getZkclientx()
    {
        // 做一下排序，保证相同的机器只使用同一个链接
        List<String> zkClusters = new ArrayList<String>(parameter.getZkClusters());
        Collections.sort(zkClusters);

        return ZooKeeperClient.getZkClient(StringUtils.join(zkClusters, ";"));
    }

    public void setAlarmHandler(AlarmHandler alarmHandler)
    {
        this.alarmHandler = alarmHandler;
    }

}
